package cn.fz.kafkademo01.service;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;

@Slf4j
@Service
public class KafkaConsumer {
    @KafkaListener(topics = "test-topic", groupId = "my-group")
    public void listen(String message) {
        System.out.println("Received Message: " + message);
    }

    @KafkaListener(topics = "test-topic", groupId = "my-group2")
    public void listen2(ConsumerRecord<String, String> record, Acknowledgment ack) {
        try {
            System.out.println("Received Message22222: " + record.value());
            // 处理消息的逻辑
            // ...
            ack.acknowledge();
        } catch (Exception e) {
            // 处理异常
            log.error("监听kafka异常：topic:{},groupId:{},error:{}", record.topic(), "aa", e.getMessage());
        }
    }

    // @KafkaListener(topics = "test-topic", groupId = "my-group3")
    // public void listen3(String message) {
    //     System.out.println("Received Message33333: " + message);
    // }
}
